package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 * 利用Flink State状态实时计算UV（访客数）
 *      UV，全称Unique Visitor 唯一访客数
 *      todo 实时统计每日用户UV，需要记录每个用户上一次访问日期，每当来一条数据时，进行对比，
 *          1. 如果日期不等于数据日期，就表示用户为新访问，uv加1
 *          2. 如果日期等于数据日期，就表示不属于今天新访问用户，uv不变（相当于uv加0）
 * @author xuanyu
 * @date 2025/10/29
 */
public class _01RealtimeUvStateDemo {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("topic-log")
			.setGroupId("realtime-uv-g1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);

		// 3. 数据转换
		/*
			todo 分析数据可知，mid为设备标识符，作为用户ID
{
  "common": {
    "ar": "230000",		-- 地区编码
    "ba": "iPhone",		-- 品牌名称
    "ch": "Appstore",	-- 渠道
    "is_new": "1",		-- 是否为新用户
    "md": "iPhone Xs Max",	-- 型号
    "mid": "mid_590764",	-- 设备ID
    "os": "iOS 12.4.1",		-- 操作系统
    "uid": "359",			-- 会员ID
    "vc": "v2.1.134"		-- 版本号
  },
  "page": {
    "during_time": 12340,
    "last_page_id": "good_detail",
    "page_id": "cart"
  },
  "ts": 1713435281000
}
		 */
		// todo topic-log中日志数据，分为启动日志start和页面日志page，要计算pv、uv等指标，属于页面日志统计指标
		//	3-1. 需要过滤出page页面日志数据，进行进一步处理
		SingleOutputStreamOperator<String> stream1 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析
					JSONObject page = JSON.parseObject(value).getJSONObject("page");
					// 比较
					return null != page ;
				}
			}
		);

		// 3-2. 按照mid分组
		KeyedStream<String, String> stream2 = stream1.keyBy(
			json -> JSON.parseObject(json).getJSONObject("common").getString("mid")
		);

		// 3-3. 比较，当前用户是否为今天首次访问
		SingleOutputStreamOperator<String> stream3 = stream2.process(
			new KeyedProcessFunction<String, String, String>() {
				// todo s1-声明状态
				private transient ValueState<String> lastVisitDateState;
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

				@Override
				public void open(Configuration parameters) throws Exception {
					// todo s2-状态初始化
					lastVisitDateState = getRuntimeContext().getState(
						new ValueStateDescriptor<>("last-visit-date", String.class)
					);
				}

				@Override
				public void processElement(String value,
				                           Context ctx,
				                           Collector<String> out) throws Exception {
					// 解析json获取ts值
					Long tsValue = JSON.parseObject(value).getLong("ts");
					// 转换日期字符串
					String visitDate = format.format(tsValue / 1000);

					// todo s3-获取状态中日期
					String lastVisitDate = lastVisitDateState.value();

					// todo s4-使用状态值（日期）
					// 当期访问日期，与以前访问日期不一样，说明此时为第一次访问
					if (!visitDate.equals(lastVisitDate)) {
						out.collect(value);
						// todo s5- 更新当前日期
						lastVisitDateState.update(visitDate);
					}
				}
			}
		);

		// 4. 数据接收器-sink
		// stream3.print("uv");
		KafkaSink<String> sink = KafkaSink.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setRecordSerializer(KafkaRecordSerializationSchema.builder()
				.setTopic("unique-visitor")
				.setValueSerializationSchema(new SimpleStringSchema())
				.build()
			)
			.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
			.build();
		stream3.sinkTo(sink);

		// 5. 触发执行-execute
		env.execute("RealtimeUvStateDemo") ;
	}

}
